package com.aispeech.task;

import com.aispeech.task.config.EtcdConfig;
import com.aispeech.task.lasr.event.LASREventSource;
import com.aispeech.task.lasr.fsm.LASRState;
import com.aispeech.task.mgr.event.TaskEvent;
import com.aispeech.task.mgr.mapper.StateMachineMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Map;

@Slf4j
public class Bootstrap {
	public static void main(String[] args) throws Exception {
		Map<String, String> conf = (Map)EtcdConfig.Instance().current();

		conf.forEach((k, v) -> { log.info("[ conf ] {} : {}", k, v); });

		final ParameterTool params = ParameterTool.fromMap(conf);

		final SourceFunction<TaskEvent> source = new LASREventSource();
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.enableCheckpointing(2000L);
		env.getConfig().setGlobalJobParameters(params);
//		env.setParallelism(1);1

		DataStream<TaskEvent> events = env.addSource(source)
			.map(new StateMachineMapper(LASRState.Initial));

		events.print();

		env.execute("lasr-task executor");
	}
}
